package cn.hyxy.hadoop;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class Demo16_DBOutputMR extends Configured implements Tool{
	@Override
	public int run(String[] args) throws Exception {
		if (args.length!=1) {
			System.out.println("Usage:in");
			return -1;
		}
		
		Configuration config=getConf();
		
		//关键是设置：driver,url,name,password
		String driver="com.mysql.jdbc.Driver";
		String url="jdbc:mysql://192.168.191.1:3306/hadoop?characterEncoding=UTF-8&useSSL=false";
		String name="root";
		String password="root";
		DBConfiguration.configureDB(config, driver, url, name, password);//有一个自己的特殊的DBCOnfiguration 
		
		Job job = Job.getInstance(config, "DBOutput");
		job.setJarByClass(getClass());
		job.setMapperClass(DBMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Stud.class);
		
		job.setReducerClass(DBReducer.class);
		job.setOutputKeyClass(Stud.class);
		job.setOutputValueClass(NullWritable.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		TextInputFormat.addInputPath(job, new Path(args[0]));
		job.setOutputFormatClass(DBOutputFormat.class);
		DBOutputFormat.setOutput(job, "studs", "id","name","age"); //关键！！！
		
		return job.waitForCompletion(true)?1:0;		
	
	}

	public static void main(String[] args) throws Exception {
		int code = ToolRunner.run(new Demo16_DBOutputMR(), args);
		System.exit(code);
	}	
	
	public static class DBMapper extends Mapper<LongWritable, Text, Text, Stud>{
		private Text key2=new Text();
		private Stud stud=new Stud();
		@Override
		public void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Stud>.Context context)
				throws IOException, InterruptedException {
			String[] strings=value.toString().split("\\s+");
			stud.setId(strings[0]);
			System.out.println(strings[0]);
			stud.setName(strings[1]);
			stud.setAge(Integer.parseInt(strings[2]));
			key2.set(strings[0]);
			context.write(key2, stud);
		}
	}
	
	public static class DBReducer extends Reducer<Text, Stud, Stud, NullWritable>{
		@Override
		public void reduce(Text key3, Iterable<Stud> value3, Reducer<Text, Stud, Stud, NullWritable>.Context context)
				throws IOException, InterruptedException {
			Stud stud=value3.iterator().next();
			
			context.write(stud, NullWritable.get());
		}
	}	
	
	public static class Stud implements Writable,DBWritable{
		private String id;
		private String name;
		private int age;

		
		@Override
		public void readFields(DataInput in) throws IOException {
			id=in.readUTF();
			name=in.readUTF();
			age=in.readInt();
		}

		@Override
		public void write(DataOutput out) throws IOException {
			out.writeUTF(id);
			out.writeUTF(name);
			out.writeInt(age);
		}
		
		@Override
		public void readFields(ResultSet resultSet) throws SQLException {
			id=resultSet.getString(1);
			name=resultSet.getString(2);
			age=resultSet.getInt(3);
		}

		@Override
		public void write(PreparedStatement ps) throws SQLException {
			ps.setString(1, id);
			ps.setString(2, name);
			ps.setInt(3, age);
		}
		
		public String getId() {
			return id;
		}

		public void setId(String id) {
			this.id = id;
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}

		public int getAge() {
			return age;
		}

		public void setAge(int age) {
			this.age = age;
		}
	}
}
